package com.bw.demo.dbiao;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.io.output.NullWriter;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
/*
 * 有多少个文本就实例化多少个实体类
 */
public class DbiaoReducer extends Reducer<Text, DbiaoBean, DbiaoBean, NullWriter> {
	@Override
	protected void reduce(Text key, Iterable<DbiaoBean> values,Context context)
			throws IOException, InterruptedException {
		DbiaoBean spBean = null;
        //用于存放次要实体类
		ArrayList<DbiaoBean> dingdans = new ArrayList<DbiaoBean>();
		try {
			for (DbiaoBean dbiaoBean : values) {
				if (dbiaoBean.getTable().equals("sp")) {
					//BeanUtils.copyProperties就是把某个bean的所有属性值复制到另一个bean中
					spBean = new DbiaoBean();
					BeanUtils.copyProperties(spBean, dbiaoBean);
				} else {
					DbiaoBean dingdan = new DbiaoBean();
                    //把某个bean的所有属性值复制到另一个bean中
					BeanUtils.copyProperties(dingdan, dbiaoBean);
					dingdans.add(dingdan);
				}
			}
            //把次要的实体类中的数据存到主要的实体类（具有唯一性的类）中
			for (DbiaoBean dbiaoBean : dingdans) {
				//把商品表缺失的属性补全
				spBean.setId(dbiaoBean.getId());
				spBean.setDate(dbiaoBean.getDate());
				spBean.setAmount(dbiaoBean.getAmount());
				context.write(spBean, null);
			}
		} catch (IllegalAccessException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InvocationTargetException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
